package com.pw.study.realtime.handle

import com.alibaba.fastjson.JSON
import com.google.gson.Gson
import com.pw.study.realtime.bean.{ActionBean, ActionLog, CommonBean, OrderBean, OrderDetail, SaleDetail, StartBean, StartBeanInfo}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import redis.clients.jedis.Jedis

import java.time.{Instant, LocalDateTime, ZoneId}
import java.time.format.DateTimeFormatter
import java.util
import scala.collection.mutable.ListBuffer

object RDDHandler {
  //两张表进行关联
  def rddToOrderInfoAndOrderDetail(ds: DStream[(String, (Option[OrderBean], Option[OrderDetail]))]): DStream[SaleDetail] = {
    println("开始两张表整合。。。。。。。。。。。。。。。。。")
    ds.mapPartitions(partition=>{
      val gson: Gson = new Gson()
      val jedis: Jedis = RedisHandler.getJedisClient()
      val saleDetails: ListBuffer[SaleDetail] = ListBuffer[SaleDetail]()
      partition.foreach{
        case(orderId,(orderInfo,orderDetail))=>{
          println(s"order=>${orderId} ==orderInfo=>${orderInfo} ==orderDetail=》 ${orderDetail}")
          //1.左边有值
          if(orderInfo != None){
            val oi: OrderBean = orderInfo.get
            if(orderDetail !=None){
              val od: OrderDetail = orderDetail.get
              val saleDetail = new SaleDetail(oi, od)
              saleDetails.append(saleDetail)
            }else{
              println("info......")
              val earlyCameOD: util.Set[String] = jedis.smembers("orderDetail:" + orderId)
              if(earlyCameOD.size()!=0){
                earlyCameOD.forEach(row=>{
                  val detail: OrderDetail = JSON.parseObject(row, classOf[OrderDetail])
                  val detail1: SaleDetail = new SaleDetail(oi, detail)
                  saleDetails.append(detail1)
                })
              }else{
                jedis.set("orderInfo:"+orderId,gson.toJson(oi))
              }
            }

          }//左边没值
          else{
            println("进入小树岭。。。")
            //
            val od: OrderDetail = orderDetail.get
            val joi: String = jedis.get("orderInfo:" + orderId)
            if(joi!=null){
              val orderInfo: OrderBean = JSON.parseObject(joi, classOf[OrderBean])
              val detail = new SaleDetail(orderInfo, od)
              saleDetails.append(detail)
            }else{
              jedis.sadd("orderDetail:"+orderId,gson.toJson(od))
            }
          }
        }
      }


      jedis.close()
      saleDetails.iterator
    })


  }

  //封装成action
  def rddToActionBean(ds2: DStream[String]): DStream[ActionLog] = {
    println("样类封装数据。。。。。。。")
    ds2.map(row => {
      val obj = JSON.parseObject(row)
      import collection.JavaConverters._
      val beans = JSON.parseArray(obj.getString("actions"), classOf[ActionBean]).asScala.toList
      val commonBean = JSON.parseObject(obj.getString("common"), classOf[CommonBean])
      val ts = obj.getLong("ts")
      val log = ActionLog(beans, commonBean, ts)
      log
    })
  }

  def rddToOrderBean(rdd: RDD[ConsumerRecord[String, String]]): RDD[OrderBean] = {
    println("数据开始处理。。。")
    rdd.map(row => {
      val bean = JSON.parseObject(row.value(), classOf[OrderBean])

      //封装时间相关的字段
      val formatter1: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")
      val formatter2: DateTimeFormatter = DateTimeFormatter.ofPattern("HH")
      //"create_time": "2022-01-17 11:17:05",
      val formatter3: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
      println(bean.create_time)
      if (bean.create_time != null) {
        val dateTime = LocalDateTime.parse(bean.create_time, formatter3)
        bean.create_date = dateTime.format(formatter1)
        bean.create_hour = dateTime.format(formatter2)
      }
      println(s"bean:+ ${bean}")
      bean
    })
  }

  def saveDataToRedis(rddRedis: RDD[StartBean]) = {
    println("redis----开始写入数据")
    rddRedis.foreachPartition(partition => {
      val jedis = RedisHandler.getJedisClient()
      partition.foreach(bean => jedis.sadd("DAU" + bean.day, bean.mid))
      jedis.close()
    })
    println("redis----写入数据结束")
  }

  def saveDataToHbase(rddRedis: RDD[StartBean]) = {
    println("----开始写入数据Hbase")
    import org.apache.phoenix.spark._
    val tableName = "gmall2022_startup"
    import org.apache.hadoop.hbase.HBaseConfiguration
    val seq = Seq("AR", "BA", "CH", "IS_NEW", "MD", "MID", "OS", "UID", "VC", "ENTRY", "LOADING_TIME", "OPEN_AD_ID", "OPEN_AD_MS", "OPEN_AD_SKIP_MS", "DAY", "HOUR", "TS")
    val conf = HBaseConfiguration.create()
    val zk = Some("hadoop112:2181")
    rddRedis.saveToPhoenix(tableName, seq, conf, zk)

    println("----写入数据结束Hbase")
  }

  def rddToDataETL(rddBean: RDD[StartBean]): RDD[StartBean] = {
    println("处理rdd..数据抽取")
    rddBean.map(bean => ((bean.mid, bean.day), bean)).groupByKey().flatMap({ case ((mid, day), bean) =>
      bean.toList.sortBy(_.ts).take(1)
    })
  }

  def rddToBean(rdd: RDD[ConsumerRecord[String, String]]): RDD[StartBean] = {
    println("处理rdd..封装为指定格式")
    rdd.map(record => {
      val map = JSON.parseObject(record.value())

      val startBean = JSON.parseObject(map.getString("common"), classOf[StartBean])
      val beanInfo = JSON.parseObject(map.getString("start"), classOf[StartBeanInfo])
      startBean.mergeDataOfStartBeanInfo(beanInfo)
      startBean.ts = map.getLong("ts")
      //封装数据
      val formatter_day = DateTimeFormatter.ofPattern("yyyy-MM-dd")
      val formatter_hour = DateTimeFormatter.ofPattern("HH")
      val dateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(startBean.ts), ZoneId.of("Asia/Shanghai"))
      startBean.day = dateTime.format(formatter_day)
      startBean.hour = dateTime.format(formatter_hour)
      startBean
    })

  }

}
